package xiaofei

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SQLContext, UserDefinedFunction}
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.functions._
import utils._

object PersonStatisInfoXf {
  var semesterYearToXQXNudf : UserDefinedFunction = _
  var getLastSemesterYearudf : UserDefinedFunction = _

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("person_static_info_xf").setMaster("yarn-client")
    val sc = new SparkContext(conf)
    val hive = new HiveContext(sc)

    hive.sql("use sunmnet_app_new")
    initTempTable(hive);
    val xslsrs = hive.sql("SELECT xsbh AS code,1 AS type FROM sp_xs_xs where sfzx=1 UNION ALL SELECT jzgh AS code,2 AS type FROM sp_jzg_jzg WHERE sfzz=1").cache
    val xfjl = hive.sql("""SELECT a.xh as code ,
                          |                substr(xnxqbh,1,9) as semester_year,
                          |                substr(xnxqbh,11,1) as semester,
                          |                d.shdz as building,
                          |                a.xfje as fee,
                          |                sum(a.xfje) OVER (partition by a.xh,substr(c.xn,1,4),c.xnxqbh) as zfee,
                          |                sum(a.xfje) OVER (partition by substr(c.xn,1,4),c.xnxqbh,b.type) as qxfee,
                          |                (CASE
                          |                    WHEN d.shlxbm=1 AND hour(a.xfsj)>=5 AND hour(a.xfsj)<=10 THEN 'zc'
                          |                    WHEN d.shlxbm=1 AND hour(a.xfsj)>10 AND hour(a.xfsj)<=15 THEN 'wc'
                          |                    WHEN d.shlxbm=1 AND hour(a.xfsj)>15 AND hour(a.xfsj)<=20 THEN 'nc'
                          |                    WHEN d.shlxbm=1 AND hour(a.xfsj)>20 THEN 'xy'
                          |                    WHEN d.shlxbm=2 THEN 'cs'
                          |                    ELSE 'qt'
                          |                    END
                          |                ) as lx,
                          |                day(a.xfsj) as days,
                          |                month(a.xfsj) as month,
                          |                b.type
                          |        FROM(
                          |			select distinct * from sp_ykt_xflsjlb) a
                          |        JOIN student_teacher b on a.xh=b.code,sp_xx_xnxq c,sp_ykt_shxx d
                          |        WHERE a.xfsj >= to_date(c.qsrq) and a.xfsj <= to_date(c.jsrq) and d.shbh = a.shbh""".stripMargin('|')).cache
    logPrinter("消费记录读取完毕")


    val xfqk = xfjl.groupBy(
      "code",
      "semester_year",
      "semester",
      "building",
      "month",
      "lx",
      "type"
    ).agg(
      sum("fee").alias("fee"),
      max("zfee").alias("zfee"),
      max("qxfee").alias("qxfee"),
      count("lx").alias("cou")
    ).cache
    logPrinter("消费情况DataFrame创建完毕")

    val xfrs = xfqk.select(
      "semester",
      "semester_year",
      "type",
      "code"
    ).distinct()
      .groupBy("semester","semester_year","type")
      .count()
      .withColumnRenamed("count","cou")
      .cache
    logPrinter("消费人数DataFrame创建完毕")

    semesterYearToXQXNudf = udf(semesterYearToXQXN _)
    getLastSemesterYearudf = udf(getLastSemesterYear _)



    insertIntoTempTable(scxfje(xfqk))
    insertIntoTempTable(sczzxfbfb(xfqk))
    insertIntoTempTable(scxftbdcnxjbfb(xfqk))
    insertIntoTempTable(scpjxf(xfqk))
    insertIntoTempTable(scpjxfcgqxpjxf(xfqk))
    insertIntoTempTable(csxfzzxfbfb(xfqk))
    insertIntoTempTable(csxfbfbtbsxnzj(xfqk))
    insertIntoTempTable(scxfzgyfmcpjzcgbyqxpjz(xfqk))
    insertIntoTempTable(scxfzdyf(xfqk))
    insertIntoTempTable(scxfzdyfmcpjz(xfqk))


    insertIntoTable(hive)
  }

  def semesterYearToXQXN(semester_year: String, semester: String): String ={
    val arr = semester_year.split('-')
    return arr(0) + arr(1) + semester
  }

  def getLastSemesterYear(semesterYear: String): String ={
    val arr = semesterYear.split('-')
    return s"${arr(0).toInt - 1}-${arr(1).toInt - 1}"
  }

  def initTempTable(hive: HiveContext): Unit = {
    hive.sql("drop table if exists person_static_info_tmp");
    hive.sql("""CREATE TABLE person_static_info_tmp(
               |		code STRING COMMENT '学号',
               |		ywcl STRING COMMENT '字段英文名称',
               |		zwcl STRING COMMENT '字段中文名称',
               |		value STRING COMMENT '字段值',
               |		sjly STRING COMMENT '数据来源',
               |		xqxn INT COMMENT '学年',
               |		lx INT COMMENT '类型：1：成绩，2：消费，3：图书馆，4：上网'
               |	)
               |	PARTITIONED BY (tjlx STRING)""".stripMargin('|'))

    hive.sql("set hive.exec.dynamic.partition=true")
    hive.sql("set hive.exec.dynamic.partition.mode=nonstrict")
    hive.sql("set hive.exec.max.dynamic.partitions.pernode=100000")
    hive.sql("set hive.exec.max.dynamic.partitions=10000000")
    hive.sql("set hive.exec.max.created.files=100000")
    logPrinter("临时表初始化完毕")



  }

  def insertIntoTempTable(data: DataFrame): Unit ={
    data.write.mode("append").partitionBy("tjlx").insertInto("person_static_info_tmp")
  }

  def insertIntoTable(hive: HiveContext): Unit ={
    hive.sql("insert overwrite table person_static_info partition(tjlx) select * from person_static_info_tmp")
    logPrinter("临时表数据插入到目标表完毕")
  }

  /**
    * c_1
    * 三餐消费金额
    * @param xfqk
    */
  def scxfje(xfqk:DataFrame): DataFrame ={
    val result = xfqk.select("code","fee","semester_year","lx","semester")
      .filter("lx in ('zc','wc','nc')")
      .groupBy("code","semester_year","semester")
      .sum("fee")
      .withColumnRenamed("sum(fee)","value")
      .withColumn("xqxn",semesterYearToXQXNudf(xfqk("semester_year"),xfqk("semester")))
      .selectExpr(
        "code",
        "'c_1' as ywcl",
        "'三餐消费金额' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_1' as tjlx"
      )
    logPrinter("三餐消费金额统计完毕")

    return result
  }

  /**
    * c_2
    * 三餐占总消费的百分比
    * @param xfqk
    */
  def sczzxfbfb(xfqk: DataFrame): DataFrame ={
    var result = xfqk.select("code","fee","semester_year","lx","semester","zfee")
      .filter("lx in ('zc','wc','nc')")
      .groupBy("code","semester_year","semester")
      .agg(sum("fee"),max("zfee"))

    result = result.withColumn("value", round(result("sum(fee)") / result("max(zfee)")))
      .withColumn("xqxn",semesterYearToXQXNudf(xfqk("semester_year"),xfqk("semester")))
      .selectExpr(
        "code",
        "'c_2' as ywcl",
        "'三餐占总消费的百分比' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_2' as tjlx"
      )

    logPrinter("三餐消费金额统计完毕，插入表格完毕")

    return result
  }

  /**
    * c_3
    * 三餐消费同比上学年增加百分比
    * @param xfqk
    */
  def scxftbdcnxjbfb(xfqk: DataFrame): DataFrame ={
    val semesterXf = xfqk.select("code","fee","semester_year","lx","semester")
      .filter("lx in ('zc','wc','nc')")
      .groupBy("code","semester_year","semester")
      .sum("fee")
      .withColumnRenamed("sum(fee)", "scfee")
      .withColumn("last_semester_year",getLastSemesterYearudf(col("semester_year")))

    val lastYearXf = semesterXf.selectExpr(
      "code",
      "scfee",
      "semester_year",
      "semester"
    )
      .selectExpr(
        "code as code2",
        "scfee as last_year_scfee",
        "semester_year as semester_year2",
        "semester as semester2"
      )

    val result = semesterXf.join(lastYearXf, semesterXf("code") === lastYearXf("code2") && semesterXf("semester") === lastYearXf("semester2") && semesterXf("last_semester_year") === lastYearXf("semester_year2"), "left")
      .withColumn("value",(col("scfee") - col("last_year_scfee")) / col("last_year_scfee") * 100)
      .withColumn("xqxn",semesterYearToXQXNudf(col("semester_year"),col("semester")))
      .selectExpr(
        "code",
        "'c_3' as ywcl",
        "'三餐消费同比上学年增加百分比' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_3' as tjlx"
    )

    logPrinter("三餐消费同比上学年增加百分比，插入表格完毕")

    return result
  }

  /**
    * c_4
    * 三餐平均消费
    * @param xfqk
    */
  def scpjxf(xfqk: DataFrame): DataFrame ={xfqk
    var result = xfqk.select("code","fee","semester_year","lx","semester","cou")
      .filter("lx in ('zc','wc','nc')")
      .groupBy("code","semester_year","semester")
      .agg(sum("fee"),sum("cou"))

    result = result.withColumn("value",result("sum(fee)") / result("sum(cou)"))
      .withColumn("xqxn",semesterYearToXQXNudf(xfqk("semester_year"),xfqk("semester")))
      .selectExpr(
        "code",
        "'c_4' as ywcl",
        "'三餐平均消费' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_4' as tjlx"
      )

    logPrinter("三餐平均消费计算完毕")

    return result;
  }

  /**
    * c_5
    * 三餐平均消费超过全校平均消费
    * @param xfqk
    */
  def scpjxfcgqxpjxf(xfqk: DataFrame): DataFrame ={
    val zxfqk = xfqk.select("code","semester","semester_year","fee","cou","lx","type")
      .filter("lx in ('zc','wc','nc')")
      .groupBy("semester","semester_year","type")
      .agg(sum("cou"), sum("fee"))
      .withColumnRenamed("sum(cou)","zcs")
      .withColumnRenamed("sum(fee)","zxf")

    val grxf = xfqk.join(zxfqk, Seq("semester", "semester_year"))
      .filter("lx in ('zc','wc','nc')")
      .groupBy("code","semester","semester_year")
      .agg(sum("fee"), sum("cou"))
      .withColumnRenamed("sum(fee)","grzxf")
      .withColumnRenamed("sum(cou)","grzcs")

    val result = grxf.join(zxfqk,Seq("semester","semester_year"),"left")
      .withColumn("value",col("grzxf") / col("grzcs") - col("zxf") / col("zcs"))
      .withColumn("xqxn",semesterYearToXQXNudf(col("semester_year"), col("semester")))
      .selectExpr(
        "code",
        "'c_5' as ywcl",
        "'三餐平均消费超过全校平均消费' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_4' as tjlx"
      )
    logPrinter("三餐平均消费超过全校平均消费计算完毕")

    return result

  }

  /**
    * c_6
    * 超市消费占总消费百分比
    *
    */
  def csxfzzxfbfb(xfqk: DataFrame): DataFrame = {
    val result = xfqk.select("code","fee","zfee", "semester_year", "semester")
      .filter("lx = 'cs'")
      .groupBy("semester_year","semester","code")
      .agg(sum("fee"), sum("zfee"))
      .withColumnRenamed("sum(fee)", "fee")
      .withColumnRenamed("sum(zfee)","zfee")
      .withColumn("value", col("fee") / col("zfee") * 100)
      .withColumn("xqxn", semesterYearToXQXNudf(col("semester_year"), col("semester")))
      .selectExpr(
        "code",
        "'c_6' as ywcl",
        "'超市消费占总消费百分比' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_6' as tjlx"
      )

    logPrinter("超市消费占总消费百分比统计完毕")

    return result;
  }

  /**
    * c_7
    * 超市消费百分比同比上学年增加
    * @param xfqk
    * @return
    */
  def csxfbfbtbsxnzj(xfqk: DataFrame): DataFrame = {
    val semesterXf = xfqk.select("code","fee","semester_year","lx","semester")
      .filter("lx = 'cs'")
      .groupBy("code","semester_year","semester")
      .sum("fee")
      .withColumnRenamed("sum(fee)", "scfee")
      .withColumn("last_semester_year",getLastSemesterYearudf(col("semester_year")))

    val lastYearXf = semesterXf.selectExpr(
      "code",
      "scfee",
      "semester_year",
      "semester"
    )
      .selectExpr(
        "code as code2",
        "scfee as last_year_scfee",
        "semester_year as semester_year2",
        "semester as semester2"
      )

    val result = semesterXf.join(lastYearXf, semesterXf("code") === lastYearXf("code2") && semesterXf("semester") === lastYearXf("semester2") && semesterXf("last_semester_year") === lastYearXf("semester_year2"), "left")
      .withColumn("value",(col("scfee") - col("last_year_scfee")) / col("last_year_scfee") * 100)
      .withColumn("xqxn",semesterYearToXQXNudf(col("semester_year"),col("semester")))
      .selectExpr(
        "code",
        "'c_7' as ywcl",
        "'超市消费百分比同比上学年增加百分比' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_7' as tjlx"
      )

    logPrinter("超市消费百分比同比上学年增加统计完毕")

    return result;

  }

  /**
    * c_8
    * 三餐消费最高月份每餐平均值超过本月全校平均值
    */
  def scxfzgyfmcpjzcgbyqxpjz(xfqk: DataFrame): DataFrame = {
    val grzxf = xfqk.filter("lx in ('zc','wc','nc')")
      .groupBy("code","month","semester","semester_year")
      .agg(sum("cou").alias("zcs"), sum("fee").alias("zxf"))
      .withColumn("pjxf",col("zxf") / col("zcs"))
      .withColumn("row_number", row_number().over(Window.partitionBy("code","semester","semester_year").orderBy(col("pjxf").desc)))
      .filter("row_number = 1")


    val qxzxf = grzxf.groupBy("month","semester","semester_year")
      .agg(sum("zcs").alias("zcs"), sum("zxf").alias("zxf"))
      .withColumn("qxpjxf",col("zxf") / col("zcs"))

    val result = grzxf.join(qxzxf,Seq("month","semester","semester_year"),"left")
      .withColumn("xqxn", semesterYearToXQXNudf(col("semester_year"), col("semester")))
      .selectExpr(
        "code",
        "'c_8' as ywcl",
        "'三餐消费最高月份每餐平均值超过本月全校平均值' as zwcl",
        "pjxf - qxpjxf as value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_8' as tjlx"
      )

    logPrinter("c_8,三餐消费最高月份每餐平均值超过本月全校平均值计算完毕")

    return result;
  }

  /**
    * c_9
    * 三餐消费最低月份
    * @param xfqk
    * @return
    */
  def scxfzdyf(xfqk: DataFrame): DataFrame = {
    val result = xfqk.filter("lx in ('zc','wc','nc')")
      .groupBy("code", "month", "semester", "semester_year")
      .agg(sum("fee").alias("fee"))
      .withColumn("row_number", row_number().over(Window.partitionBy("code","semester","semester_year").orderBy(col("fee"))))
      .filter("row_number = 1")
      .withColumn("xqxn", semesterYearToXQXNudf(col("semester_year"), col("semester")))
      .selectExpr(
        "code",
        "'c_9' as ywcl",
        "'三餐消费最低月份' as zwcl",
        "month as value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_9' as tjlx"
      )

    logPrinter("三餐消费最低月份统计完毕")
    return result;
  }

  /**
    * c_10
    * 三餐消费最低月份每餐平均值
    * @param xfqk
    * @return
    */
  def scxfzdyfmcpjz(xfqk: DataFrame): DataFrame = {
    val month = xfqk.filter("lx in ('zc','wc','nc')")
      .groupBy("code", "month", "semester", "semester_year")
      .agg(sum("fee").alias("fee"))
      .withColumn("row_number", row_number().over(Window.partitionBy("code","semester","semester_year").orderBy(col("fee"))))
      .filter("row_number = 1")

    val result = xfqk.filter("lx in ('zc','wc','nc')")
      .groupBy("code", "month", "semester", "semester_year")
      .agg(sum("fee").alias("fee"),sum("cou").alias("cou"))
      .join(month, Seq("code", "month", "semester", "semester_year"), "inner")
      .withColumn("value", col("fee") / col("cou"))
      .withColumn("xqxn", semesterYearToXQXNudf(col("semester_year"), col("semester")))
      .selectExpr(
        "code",
        "'c_10' as ywcl",
        "'三餐消费最低月份每餐平均值' as zwcl",
        "value",
        "'sp_ykt_xflsjlb' as sjly",
        "xqxn",
        "2 as lx",
        "'c_10' as tjlx"
      )

    logPrinter("三餐消费最低月份每餐平均值统计完毕")
    return result
  }
}
